package core

import (
	"context"
	"encoding/json"
	"github.com/gin-gonic/gin"
	"github.com/gorilla/websocket"
	uuid "github.com/satori/go.uuid"
	"suitbim.com/media/utils"
)

var (
	dispatcher *Dispatcher
)

// Dispatcher 管理端的主结构体，职责：
// 1. 维护与Dispatcher（remote media center）之间的信号来往（websocket协议）
// 2. 维护负责给Dispatcher推流（tcp）的client列表
type Dispatcher struct {
	channels          []Channel
	ChannelMap        map[string]*Channel
	ctx               context.Context
	cancel            context.CancelFunc
	viewersManager    *WSServer
	signalManager     *WSServer
	ptzManager        *WSServer // 管理ptz结果指令
	SignalConnections []*websocket.Conn
}

func NewDispatcher() *Dispatcher {
	disp := Dispatcher{}
	disp.SignalConnections = make([]*websocket.Conn, 0)
	disp.channels = LoadChannels()
	disp.ChannelMap = make(map[string]*Channel)
	for i := 0; i < len(disp.channels); i++ {
		disp.ChannelMap[disp.channels[i].Code] = &disp.channels[i]
	}
	disp.ctx, disp.cancel = context.WithCancel(context.Background())
	dispatcher = &disp
	return dispatcher
}

func (disp *Dispatcher) Start() {
	disp.signalManager = &WSServer{Port: G.Dispatcher.Signal.Port}
	disp.viewersManager = &WSServer{Port: G.Dispatcher.Viewer.Port}
	streamReceiver := TCPServer{Port: G.Dispatcher.Stream.Port}
	captureReceiver := TCPServer{Port: G.Dispatcher.Capture.Port}
	go streamReceiver.Start("Stream", disp.ctx, disp, OnStreamPusherEnter)    // 开启tcp接收pusher推的媒体流，例如rtsp流等持续性流
	go captureReceiver.Start("Capture", disp.ctx, disp, OnCapturePusherEnter) // 开启tcp接收pusher推的文件流，例如截图等短暂流
	go disp.signalManager.Start("Signal", disp.OnPusherManagerEnter)          // 开启服务端与所有pusher客户端的websocket通信服务
	go disp.viewersManager.Start("Viewer", disp.OnViewerEnter)                // 开启前端与websocket通信服务
	disp.StartHttpServer()                                                    // 开启与前端交互的http接口
}

// OnViewerEnter 处理来自前端流消费者的事件
func (disp *Dispatcher) OnViewerEnter(c *gin.Context) {
	// 升级get请求为websocket协议
	ws, err := upgrader.Upgrade(c.Writer, c.Request, nil)
	if err != nil {
		return
	}

	chCode := c.Query("ch")
	G.Log.Infof("Viewer[%s] require channel [%s]", c.ClientIP(), chCode)
	ch, ok := disp.ChannelMap["ch"+chCode]
	var v *Viewer
	if ok {
		// 将viewer添加值dispatch的Clients列表中(登记)
		v = &Viewer{conn: ws, vid: uuid.Must(uuid.NewV4(), nil).String()}
		ch.viewers = append(ch.viewers, v)
		G.Log.Infof("Channel[%s] remain [%d] viewers", ch.Code, len(ch.viewers))
		G.Log.Infof("Channel[%s] holding Pusher[%v]", ch.Code, ch.pusher)
		if ch.pusher == nil {
			// 如果该通道目前还没有Pusher提供服务
			// 调用信号管理员发送推流消息给远端的Pusher管理员
			disp.SendSignal(ch, &Message{Sign: SIG_OPEN, Chan: ch.Code}, 1)
		}
	} else {
		// 对于不在已配置的通道范围内的viewer，直接断开连接，避免浪费服务器资源
		_ = ws.Close()
		return
	}
	// 最后需要给浏览器正常的响应（OK，101）；
	// 否则浏览器得到的Response Status会是Finished，可能会导致无限重发
	_ = ws.WriteMessage(1, []byte("connected"))

	// 监听Viewer断开连接
	if _, _, err := ws.ReadMessage(); err != nil {
		G.Log.Errorf("Viewer[%s][%s] occur error [%v]", chCode, c.ClientIP(), err)
		ch.OnViewerLeaved(v, err)
	}
}

// OnPusherManagerEnter 处理与Pusher之间的信号交互
func (disp *Dispatcher) OnPusherManagerEnter(c *gin.Context) {
	// 升级get请求为websocket协议
	ws, err := upgrader.Upgrade(c.Writer, c.Request, nil)
	if err != nil {
		return
	}
	//disp.signalManager.conn = ws
	disp.SignalConnections = append(disp.SignalConnections, ws)
	go disp.ReceiveSignal(ws)
}

// ReceiveSignal 接收消息
func (disp *Dispatcher) ReceiveSignal(ws *websocket.Conn) {
	defer func() {
		err := recover()
		if err != nil {
			ws.Close()
		}
	}()
	for {
		_, p, err := ws.ReadMessage()
		if err != nil {
			ws.Close()
			return
		}
		msg := Message{}
		json.Unmarshal(p, &msg)
		G.Log.Infof("Received pusher message %s", string(p))
		switch msg.Sign {
		case SIG_PTZ_END:
			ch := ptzChan[msg.Chan[2:]]
			ch <- msg.Content
		}
	}
}

// SendSignal 发送消息给Pusher Managers
func (disp *Dispatcher) SendSignal(channel *Channel, msg *Message, times int) {
	G.Log.Debugf("Channel[%s] Notify Pusher Manager do [%s] [%d]Times: ", channel.Code, msg.Sign, times)
	// 通知所有已跟Dispatcher取得联系的pusher manager
	for _, remotePusherManager := range disp.SignalConnections {
		if err := remotePusherManager.WriteMessage(
			websocket.BinaryMessage,
			[]byte(utils.Struct2JsonString(*msg)),
		); err != nil {
			G.Log.Error("Notify Pusher Manager Error: ", err)
		}
	}

	// 如果是关闭信号，可以直接中断推流连接，对端检测到错误后主动停止推流动作
	if msg.Sign == SIG_CLOS && channel.pusher != nil {
		_ = channel.pusher.Close()
	}
}

// StartHttpServer 开启http服务，用来处理前端的http请求
func (disp *Dispatcher) StartHttpServer() {
	mode := G.App.Mode
	port := G.Dispatcher.Business.Port
	httpRouter := HttpRouter{}
	routerMap := httpRouter.BuildRouterMap()
	httpServer := NewHttpServer(mode)
	httpServer.StartNoRouterGroup("/rest", routerMap)
	httpServer.Run(port)
}
